package snpi;

import harness.Queue;
import harness.Random;
import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReferenceArray;

/**
 *
 * @author Trevor Brown (enqueue() and dequeue() ported from C code by Hillel Avni)
 */
public class SNPIQueue<E> implements Queue {
    
    private static final long SENTINEL = (0x0FFFFFFFL << 32) | 0xFFFFFFFFL;
    private final int NUMBER_OF_THREADS = 128;
    private final int QA_SIZE;
    
    // head and tail of queue
    private final SNPI head;
    private final SNPI tail;
    
    private final long[] publicHeads = new long[NUMBER_OF_THREADS];
    private final long[] publicTails = new long[NUMBER_OF_THREADS];
    private final AtomicReferenceArray[] qa;
    
    private final AtomicInteger nextThreadId = new AtomicInteger(0);
    private ThreadLocal<Integer> threadId = new ThreadLocal<Integer>() {
        protected Integer initialValue() {
            int index = nextThreadId.getAndIncrement();
            return index;
        }
    };
    private ThreadLocal<Integer> qaEntry = new ThreadLocal<Integer>() {
        protected Integer initialValue() {
            return threadId.get() % NUMBER_OF_THREADS;
        }
    };
    
    private ThreadLocal<Long> QAT = new ThreadLocal<Long>() {
        protected Long initialValue() { return 0L; }
    };
    
    private ThreadLocal<Long> QAH = new ThreadLocal<Long>() {
        protected Long initialValue() { return 1L; }
    };
    
    public SNPIQueue(SNPIFactory factory, int capacity) {
        if (capacity < 1) throw new RuntimeException("capacity must be at least 1");
        
        QA_SIZE = capacity+1;
        qa = new AtomicReferenceArray[QA_SIZE];
        
        // create SNPI objects for head and tail
        head = factory.newInstance(); //new SNPICASOnce();
        tail = factory.newInstance(); //new SNPICASOnce();
        
        head.increment(0);
        
        for (int i=0;i<NUMBER_OF_THREADS;++i) {
            publicHeads[i] = SENTINEL;
            publicTails[i] = SENTINEL;
        }
        for (int i=0;i<QA_SIZE;i++) {
            qa[i] = new AtomicReferenceArray(NUMBER_OF_THREADS);
        }
    }
    
    public boolean offer(final E element) {
        final boolean result;
        long h, t;
        
        if (element == null) throw new RuntimeException("cannot enqueue null");
        
        do {
            h = head.get();
            publicHeads[threadId.get()] = h;
        } while (h != head.get());
        
        if (h >= QAT.get() + QA_SIZE) {
            long tp;
            t = tail.get();
            for (int i=0;i<NUMBER_OF_THREADS;++i) {
                tp = publicTails[i];
                if (tp < t) {
                    t = tp;
                }
            }
            QAT.set(t);
        }
        
        if (h < QAT.get() + QA_SIZE) {
            head.increment(threadId.get());
            qa[(int) (h % QA_SIZE)].set(threadId.get(), element);
            result = true;
        } else {
            result = false;
        }
        
        publicHeads[threadId.get()] = SENTINEL;
        
        return result;
    }
    
    public E poll() {
        E result = null;
        long h, t;
        
        do {
            t = tail.get();
            publicTails[threadId.get()] = t;
        } while (t != tail.get());
        
        h = head.get();
        
        while (result == null) {
            if (QAH.get() > t) {
                final AtomicReferenceArray qaPart = qa[(int) (t % QA_SIZE)];
                Object val;
                
                /** DO ONE LOOP STARTING AT qaEntry AND WRAPPING AROUND **/
                for (int i=qaEntry.get();i<NUMBER_OF_THREADS;++i) {
                    val = qaPart.get(i);
                    if (val != null) {
                        if (qaPart.compareAndSet(i, val, null)) {
                            result = (E) val;
                            break;
                        }
                    }
                }
                for (int i=0;i<qaEntry.get();++i) {
                    val = qaPart.get(i);
                    if (val != null) {
                        if (qaPart.compareAndSet(i, val, null)) {
                            result = (E) val;
                            break;
                        }
                    }
                }
                /** END LOOP **/

                if (result == null) {
                    tail.increment(threadId.get());
                    t = tail.get();
                }
            } else {
                if (QAH.get() != h) {
                    QAH.set(h);
                } else {
                    break;
                }
                
                long hp;
                for (int i=0;i<NUMBER_OF_THREADS;++i) {
                    hp = publicHeads[i];
                    if (hp < h) {
                        QAH.set(hp);
                    }
                }
            }
        }
        
        publicTails[threadId.get()] = SENTINEL;
        
        return result;
    }

    public Object poll(int threadId, Random rng) {
        return poll();
    }

    public boolean offer(Object e, int threadId, Random rng) {
        return offer((E) e);
    }
    
}
